package com.ibest.base.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.github.shyiko.mysql.binlog.event.EventType;
import com.ibest.base.config.binlog.BinLogItem;
import com.ibest.base.config.binlog.Colum;
import com.ibest.base.constant.CacheConstant;
import com.ibest.base.entity.CanalBinlogBean;
import com.ibest.base.entity.TopicEntity;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.*;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;

@Service
@Slf4j
public class RedisService {

    @Resource
    private RedissonClient redissonClient;

    public void setCanalBinlogBean(CanalBinlogBean canalBinlogBean){
        String dbName=canalBinlogBean.getDatabase();
        String tableName=canalBinlogBean.getTable();
        List<String> pks=canalBinlogBean.getPkNames();
        List<JSONObject> jsonObjectList=canalBinlogBean.getData();

        jsonObjectList.stream().forEach(item->{
            StringBuffer idKey=new StringBuffer();
            pks.stream().forEach(item2->{
                idKey.append(CacheConstant.SEP_KEY);
                idKey.append(item.getString(item2));
            });
            RMap<String,Object> map=redissonClient.getMap(CacheConstant.DB_NAME_PREFIX+dbName+tableName);
            map.put(idKey.toString(),item);
        });
    }

    public Object getCanalBinlogBean(String dbName,String tableName,List<String> ids){
        RMap<String,Object> map=redissonClient.getMap(CacheConstant.DB_NAME_PREFIX+dbName+tableName);
        StringBuffer idKey=new StringBuffer();
        ids.stream().forEach(item2->{
            idKey.append(CacheConstant.SEP_KEY);
            idKey.append(item2);
        });
        return map.get(idKey.toString());
    }
    public void delCanalBinlogBean(CanalBinlogBean canalBinlogBean){
        String dbName=canalBinlogBean.getDatabase();
        String tableName=canalBinlogBean.getTable();
        List<String> pks=canalBinlogBean.getPkNames();
        List<JSONObject> jsonObjectList=canalBinlogBean.getData();

        jsonObjectList.stream().forEach(item->{
            StringBuffer idKey=new StringBuffer();
            pks.stream().forEach(item2->{
                idKey.append(CacheConstant.SEP_KEY);
                idKey.append(item.getString(item2));
            });
            RMap<String,Object> map=redissonClient.getMap(CacheConstant.DB_NAME_PREFIX+dbName+tableName);
            map.remove(idKey.toString());
        });
    }

    public void sendMsgTopic(CanalBinlogBean canalBinlogBean){
        RLocalCachedMap<String,Object> localCachedMap=redissonClient.getLocalCachedMap(CacheConstant.LOCAL_CACHE_TOPIC_KEY, LocalCachedMapOptions.defaults());
        for(Map.Entry en: localCachedMap.entrySet()){
            String topicName= (String) en.getKey();
            TopicEntity topicEntity=(TopicEntity)en.getValue();
            Set<String> tables=topicEntity.getTables();
           if(canalBinlogBean!=null&&topicEntity.getDbName().equals(canalBinlogBean.getDatabase())&&tables.contains(canalBinlogBean.getTable())){
               RTopic rTopic=redissonClient.getTopic(topicName);
               rTopic.publish(canalBinlogBean);
           }
        }
    }

    public void sendBinLogItem(BinLogItem binLogItem){
        CanalBinlogBean canalBinlogBean=new CanalBinlogBean();
        canalBinlogBean.setDatabase(binLogItem.getDatabase());
        canalBinlogBean.setTable(binLogItem.getTable());
        switch (binLogItem.getEventType()){
            case EXT_WRITE_ROWS:
                Map<String, Object> after=binLogItem.getAfter();
                canalBinlogBean.setData(Arrays.asList(new JSONObject(after)));
                canalBinlogBean.setType("INSERT");
                break;
            case EXT_UPDATE_ROWS:
                Map<String, Object> after1=binLogItem.getAfter();
                canalBinlogBean.setData(Arrays.asList(new JSONObject(after1)));
                canalBinlogBean.setType("UPDATE");
                break;
            case EXT_DELETE_ROWS:
                Map<String, Object> before=binLogItem.getBefore();
                canalBinlogBean.setData(Arrays.asList(new JSONObject(before)));
                canalBinlogBean.setType("DELETE");
                break;
        }
        Map<String, Colum> columMap=binLogItem.getColums();
        List<String> ids=new ArrayList<>();
        for(String key:columMap.keySet()){
            if(columMap.get(key).isPk){
                ids.add(key);
            }
        }
        canalBinlogBean.setPkNames(ids);
        log.info("监听到的binlog.canalBinlogBean:{}",canalBinlogBean);
        RTopic rTopic=redissonClient.getTopic(binLogItem.getDatabase());
        rTopic.publish(canalBinlogBean);
    }
}
